package com.lizhiyu.flink.demo7_state;

import com.lizhiyu.flink.demo5_window.CustomSource3;
import com.lizhiyu.flink.model.VideoOrder;
import org.apache.commons.collections.IteratorUtils;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.functions.KeySelector;
import org.apache.flink.api.java.tuple.Tuple3;
import org.apache.flink.contrib.streaming.state.EmbeddedRocksDBStateBackend;
import org.apache.flink.runtime.state.hashmap.HashMapStateBackend;
import org.apache.flink.runtime.state.storage.FileSystemCheckpointStorage;
import org.apache.flink.runtime.state.storage.JobManagerCheckpointStorage;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.util.List;
import java.util.stream.Collectors;

/**
 *   无状态计算： 同个数据进到算子里面多少次，都是一样的输出，比如 filter
 *   有状态计算：需要考虑历史状态，同个输入会有不同的输出，比如sum、reduce聚合操作
 *
 *   状态管理，可以任务是flink提供了一些工具类，保障状态存储(例如sum求和，获得的结果就是一种状态)
 *
 *   状态管理分类
 *      ManagedState （用的多）
 *          Keyed State 键控状态（用的多）
 *              一般是用richFlatFunction,或者其他richfunction里面，在open()声明周期里面进行初始化
 *              ValueState、ListState、MapState等数据结构
 *          Operator State 算子状态（用的少,部分source会用）
 *      RawState （用的少）用户自己管理和维护
 *
 *  State数据结构（状态值可能存在内存、磁盘、DB或者其他分布式存储中）
 *  ValueState 简单的存储一个值（ThreadLocal / String）
 *      ValueState.value()
 *      ValueState.update(T value)
 *  ListState 列表
 *      ListState.add(T value)
 *      ListState.get() //得到一个Iterator
 *  MapState 映射类型
 *      MapState.get(key)
 *      MapState.put(key, value)
 *
 *  状态存储位置设置
 *  （新版）
 *      HashMapStateBackend(堆中)、EmbeddedRocksDBStateBackend (非关系数据库)
 *      如果没有其他配置，系统将使用 HashMapStateBackend。
 *  (旧版）
 *      MemoryStateBackend、FsStateBackend、RocksDBStateBackend
 *      如果不设置，默认使用 MemoryStateBackend。
 */
public class State {
    public static void main(String[] args) throws Exception {
        //配置flink的状态存储方案
        //代码配置一
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setStateBackend(new HashMapStateBackend());
        env.getCheckpointConfig().setCheckpointStorage(new JobManagerCheckpointStorage());
        //代码配置二
        StreamExecutionEnvironment env2 = StreamExecutionEnvironment.getExecutionEnvironment();
        env2.setStateBackend(new EmbeddedRocksDBStateBackend());
        env2.getCheckpointConfig().setCheckpointStorage("file:///checkpoint-dir");
        //或者 env2.getCheckpointConfig().setCheckpointStorage(new FileSystemCheckpointStorage("file:///checkpoint-dir"));

    }
}
